<html>
 <head>
  <meta charset="UTF-8">
 </head>
 <body>
  <h1 data-lake-id="AlK2D" id="AlK2D"><span data-lake-id="ub08c8301" id="ub08c8301">典型回答</span></h1>
  <p data-lake-id="u5ce8b0b9" id="u5ce8b0b9"><br></p>
  <p data-lake-id="ufb7a5131" id="ufb7a5131"><span data-lake-id="u9fd39826" id="u9fd39826">Kafka的消息是存储在指定的topic中的某个partition中的。并且一个topic是可以有多个partition的。同一个partition中的消息是有序的，但是跨partition，或者跨topic的消息就是无序的了。</span></p>
  <p data-lake-id="ud4975923" id="ud4975923"><span data-lake-id="u096fceed" id="u096fceed">​</span><br></p>
  <p data-lake-id="uea84fbcf" id="uea84fbcf"><strong><span data-lake-id="u1a578dc1" id="u1a578dc1">为什么同一个partition的消息是有序的？</span></strong></p>
  <p data-lake-id="ub9c8e2e1" id="ub9c8e2e1"><span data-lake-id="u4ddc97fa" id="u4ddc97fa">​</span><br></p>
  <p data-lake-id="u0317ffb8" id="u0317ffb8"><span data-lake-id="u8984c492" id="u8984c492">因为当生产者向某个partition发送消息时，消息会被追加到该partition的日志文件（log）中，并且被分配一个唯一的 offset，文件的读写是有顺序的。而消费者在从该分区消费消息时，会从该分区的最早 offset 开始逐个读取消息，保证了消息的顺序性。</span></p>
  <p data-lake-id="u8257f4f1" id="u8257f4f1"><span data-lake-id="ubfb7485b" id="ubfb7485b">​</span><br></p>
  <p data-lake-id="u749f08b9" id="u749f08b9"><strong><span data-lake-id="u210b2cba" id="u210b2cba">基于此，想要实现消息的顺序消费，可以有以下几个办法：</span></strong></p>
  <p data-lake-id="u7ff9add3" id="u7ff9add3"><span data-lake-id="uea63baa1" id="uea63baa1">​</span><br></p>
  <p data-lake-id="u03db2730" id="u03db2730"><span data-lake-id="u1c3fb335" id="u1c3fb335">1、在一个topic中，只创建一个partition，这样这个topic下的消息都会按照顺序保存在同一个partition中，这就保证了消息的顺序消费。</span></p>
  <p data-lake-id="u8d4b7c14" id="u8d4b7c14"><span data-lake-id="u58dfab83" id="u58dfab83">​</span><br></p>
  <p data-lake-id="uaaa7ea9c" id="uaaa7ea9c"><span data-lake-id="u2dfac3e9" id="u2dfac3e9">2、发送消息的时候指定partition，如果一个topic下有多个partition，那么我们可以把需要保证顺序的消息都发送到同一个partition中，这样也能做到顺序消费。</span></p>
  <p data-lake-id="u8c18d6c9" id="u8c18d6c9"><span data-lake-id="u3b94acc7" id="u3b94acc7">​</span><br></p>
  <h1 data-lake-id="IcJzB" id="IcJzB"><span data-lake-id="uead82868" id="uead82868">扩展知识</span></h1>
  <p data-lake-id="u6060abbe" id="u6060abbe"><br></p>
  <h2 data-lake-id="XZvSh" id="XZvSh"><span data-lake-id="uc932239b" id="uc932239b">如何发到同一个partition</span></h2>
  <p data-lake-id="ub2015f7a" id="ub2015f7a"><span data-lake-id="u80adbe62" id="u80adbe62">​</span><br></p>
  <p data-lake-id="ud96abe22" id="ud96abe22"><span data-lake-id="u3e186e42" id="u3e186e42">当我们发送消息的时候，如果key为null，那么Kafka 默认采用 Round-robin 策略，也就是轮转，实现类是 DefaultPartitioner。那么如果想要指定他发送到某个partition的话，有以下三个方式：</span></p>
  <p data-lake-id="u34a98515" id="u34a98515"><span data-lake-id="uc8bbc39f" id="uc8bbc39f">​</span><br></p>
  <h3 data-lake-id="Zlt0z" id="Zlt0z"><span data-lake-id="ubbcf8daf" id="ubbcf8daf">指定partition</span></h3>
  <p data-lake-id="udd0f771a" id="udd0f771a"><br></p>
  <p data-lake-id="u41680d9b" id="u41680d9b"><span data-lake-id="u5989e730" id="u5989e730">我们可以在发送消息的时候，可以直接在ProducerRecord中指定partition</span></p>
  <p data-lake-id="u852f0c2f" id="u852f0c2f"><span data-lake-id="u28f79f53" id="u28f79f53">​</span><br></p>
  <pre lang="java"><code>
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        
        // 创建Kafka生产者
        Producer&lt;String, String&gt; producer = new KafkaProducer&lt;&gt;(getProperties());

        String topic = "hollis_topic"; // 指定要发送消息的主题
        String message = "Hello World!"; // 要发送的消息内容
        int partition = 0; // 要发送消息的分区

        // 创建包含分区信息的ProducerRecord
        ProducerRecord&lt;String, String&gt; record = new ProducerRecord&lt;&gt;(topic, partition, null, message);

        // 发送消息
        producer.send(record);

        // 关闭Kafka生产者
        producer.close();
    }
}

</code></pre>
  <h3 data-lake-id="Tj2G1" id="Tj2G1"><span data-lake-id="u93fc84ef" id="u93fc84ef">指定key</span></h3>
  <p data-lake-id="u015b2d46" id="u015b2d46"><br></p>
  <p data-lake-id="u90412e7e" id="u90412e7e"><span data-lake-id="u8e976ca5" id="u8e976ca5">在没有指定 Partition(null 值) 时, 如果有 Key, Kafka 将依据 Key 做hash来计算出一个 Partition 编号来。如果key相同，那么也能分到同一个partition中：</span></p>
  <p data-lake-id="u207b8b63" id="u207b8b63"><span data-lake-id="u9896a5d6" id="u9896a5d6">​</span><br></p>
  <pre lang="java"><code>
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        
        // 创建Kafka生产者
        Producer&lt;String, String&gt; producer = new KafkaProducer&lt;&gt;(getProperties());

        String topic = "hollis_topic"; // 指定要发送消息的主题
        String message = "Hello World!"; // 要发送的消息内容
        String key = "Hollis_key"; // 要发送消息的key

       // 创建ProducerRecord，指定主题、键和消息内容
        ProducerRecord&lt;String, String&gt; record = new ProducerRecord&lt;&gt;(topic, null, key, message);

        // 发送消息
        producer.send(record);

        // 关闭Kafka生产者
        producer.close();
    }
}

</code></pre>
  <p data-lake-id="u659661d5" id="u659661d5"><br></p>
  <h3 data-lake-id="BRlcL" id="BRlcL"><span data-lake-id="u748ec542" id="u748ec542">自定义Partitioner</span></h3>
  <p data-lake-id="ua86c5501" id="ua86c5501"><br></p>
  <p data-lake-id="u0a96e1cb" id="u0a96e1cb"><span data-lake-id="u49602961" id="u49602961">除了以上两种方式，我们还可以实现自己的分区器（Partitioner）来指定消息发送到特定的分区。</span></p>
  <p data-lake-id="uf519dbaf" id="uf519dbaf"><span data-lake-id="uf91fbdde" id="uf91fbdde">​</span><br></p>
  <p data-lake-id="udfe497e2" id="udfe497e2"><span data-lake-id="ue8f6dcd7" id="ue8f6dcd7">我们需要创建一个类实现Partitioner接口，并且重写partition()方法。</span></p>
  <p data-lake-id="u2ac324cb" id="u2ac324cb"><br></p>
  <pre lang="java"><code>
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.record.InvalidRecordException;
import org.apache.kafka.common.utils.Utils;

import java.util.List;
import java.util.Map;

public class CustomPartitioner implements Partitioner {

    @Override
    public void configure(Map&lt;String, ?&gt; configs) {
        // 可以在这里处理和获取分区器的配置参数
    }

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List&lt;PartitionInfo&gt; partitions = cluster.partitionsForTopic(topic);
        int numPartitions = partitions.size();

        if (keyBytes == null || !(key instanceof String)) {
            throw new InvalidRecordException("键不能为空且必须是字符串类型");
        }

        // 根据自定义的逻辑，确定消息应该发送到哪个分区
        String keyValue = (String) key;
        int partition = Math.abs(keyValue.hashCode()) % numPartitions;

        // 返回分区编号
        return partition;
    }

    @Override
    public void close() {
        // 可以在这里进行一些清理操作
    }
}

</code></pre>
  <p data-lake-id="ub6dac7ff" id="ub6dac7ff"><br></p>
  <p data-lake-id="uf2992913" id="uf2992913"><span data-lake-id="udae24afa" id="udae24afa">在partition()方法中，我们使用了一个简单的逻辑，根据键的哈希值将消息发送到相应的分区。为了在Kafka生产者中使用自定义的分区器，你需要在生产者的配置中指定分区器类：</span></p>
  <p data-lake-id="u9ac19f67" id="u9ac19f67"><span data-lake-id="ubf796d61" id="ubf796d61">​</span><br></p>
  <pre lang="java"><code>
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        // 设置Kafka生产者的配置属性
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("partitioner.class", "com.hollis.CustomPartitioner"); // 指定自定义分区器类

        // 创建Kafka生产者
        Producer&lt;String, String&gt; producer = new org.apache.kafka.clients.producer.KafkaProducer&lt;&gt;(props);

        String topic = "hollis_topic"; // 指定要发送消息的主题
        String message = "Hello World!"; // 要发送的消息内容
        String key = "Hollis_key"; // 要发送消息的key

        // 创建ProducerRecord，指定主题、键和消息内容
        ProducerRecord&lt;String, String&gt; record = new ProducerRecord&lt;&gt;(topic, key, message);

        // 发送消息
        producer.send(record);

        // 关闭Kafka生产者
        producer.close();
    }
}

</code></pre>
 </body>
</html>